package com.example.springboot.batch.config;

import com.example.springboot.batch.entity.User;
import com.example.springboot.batch.util.UUIDUtil;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.core.launch.support.SimpleJobLauncher;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.support.JobRepositoryFactoryBean;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.database.BeanPropertyItemSqlParameterSourceProvider;
import org.springframework.batch.item.database.JdbcBatchItemWriter;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.mapping.BeanWrapperFieldSetMapper;
import org.springframework.batch.item.file.mapping.DefaultLineMapper;
import org.springframework.batch.item.file.transform.DelimitedLineTokenizer;
import org.springframework.batch.item.validator.Validator;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.ClassPathResource;
import org.springframework.transaction.PlatformTransactionManager;

import javax.sql.DataSource;

/**
 * @desc
 * @Author wangsh
 * @date 2018/5/14 21:49
 */
@Configuration
//开启批处理支持
//@EnableBatchProcessing
public class TriggerBatchConfig {


	@Bean
	public JobRepository jobRepository(DataSource dataSource, PlatformTransactionManager transactionManager) throws Exception {
		JobRepositoryFactoryBean jobRepositoryFactoryBean = new JobRepositoryFactoryBean();
		jobRepositoryFactoryBean.setDataSource(dataSource);
		jobRepositoryFactoryBean.setTransactionManager(transactionManager);
		jobRepositoryFactoryBean.setDatabaseType("mysql");
		return jobRepositoryFactoryBean.getObject();
	}


	@Bean
	public SimpleJobLauncher jobLauncher(DataSource dataSource, PlatformTransactionManager transactionManager)
			throws Exception {
		SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
		jobLauncher.setJobRepository(jobRepository(dataSource, transactionManager));
		return jobLauncher;
	}


	@Bean
	public Step step1(StepBuilderFactory stepBuilderFactory, ItemReader<User> reader, ItemWriter<User> writer,
					  ItemProcessor<User, User> processor) {
		return stepBuilderFactory
				.get("step1")
				.<User, User>chunk(65000) //批处理每次提交65000条数据
				.reader(reader) //给step绑定reader
				.processor(processor) //给step绑定Processor
				.writer(writer) //给step绑定writer
				.build();
	}


	/**
	 * 通过文件路径读取文件内容
	 *
	 * @param pathToFile
	 * @return
	 * @throws Exception
	 */
	@Bean
	@StepScope
	public FlatFileItemReader<User> reader(@Value("#{jobParameters['input.file.name']}") String pathToFile) throws Exception {
		FlatFileItemReader<User> reader = new FlatFileItemReader<User>(); //
		reader.setResource(new ClassPathResource(pathToFile)); //
		reader.setLineMapper(new DefaultLineMapper<User>() {{ //
			setLineTokenizer(new DelimitedLineTokenizer() {{
				setNames(new String[]{"username", "password", "email", "useable"});
			}});
			setFieldSetMapper(new BeanWrapperFieldSetMapper<User>() {{
				setTargetType(User.class);
			}});
		}});

		return reader;
	}


	@Bean
	public ItemWriter<User> writer(DataSource dataSource) {//Spring能让容器中已有的Bean以参数的形式注入，Spring boot已经定义了DataSource
		JdbcBatchItemWriter<User> writer = new JdbcBatchItemWriter<User>(); //使用JDBC批处理的JdbcBatchItemWriter来写数据到数据库
		writer.setItemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<User>());
		String sql = "insert into User " + "(id,username,password,email,useable) "
				+ "values(" + UUIDUtil.getUUID() + ", :username, :password, :email,:useable)";
		writer.setSql(sql); //在此设置要执行批处理的sql语句
		writer.setDataSource(dataSource);
		return writer;
	}


	/**
	 * 绑定任务执行监听器
	 *
	 * @param jobs
	 * @param s1
	 * @return
	 */
	@Bean
	public Job importJob(JobBuilderFactory jobs, Step s1) {
		return jobs.get("importJob")
				.incrementer(new RunIdIncrementer())
				.flow(s1) //指定step
				.end()
				.listener(batchJobListener()) //绑定监听器
				.build();
	}

	@Bean
	public BatchJobListener batchJobListener() {
		return new BatchJobListener();
	}


	/**
	 * @return
	 * @desc 数据处理校验，重写processor()，自定义校验器并添加到校验中。
	 * @Author wangsh
	 * @date 2018/5/14 22:07
	 */
	@Bean
	public ItemProcessor<User, User> processor() {
		//使用自定义的ItemProcessor的实现
		BatchItemProcessor processor = new BatchItemProcessor();
		//为Processor指定自定义校验器
		processor.setValidator(batchBeanValidator());
		return processor;
	}

	@Bean
	public Validator<User> batchBeanValidator() {
		return new BatchBeanValidator<User>();
	}
}
